package com.shujia.custom.udtf;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.lazy.LazyString;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.io.Text;

import java.sql.Array;
import java.util.ArrayList;

public class UDTFDemo1 extends GenericUDTF {

    /*
        设置输出一条数据中的每一列的列名以及列的数据类型
     */
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        ArrayList<String> colNameList = new ArrayList<>();
        colNameList.add("id");
        colNameList.add("name");
        colNameList.add("cardId");

        ArrayList<ObjectInspector> colTypeList = new ArrayList<>();
        colTypeList.add(PrimitiveObjectInspectorFactory.javaLongObjectInspector);
        colTypeList.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        colTypeList.add(PrimitiveObjectInspectorFactory.javaLongObjectInspector);

        return ObjectInspectorFactory.getStandardStructObjectInspector(colNameList, colTypeList);
    }

    /*
        调用函数时需要执行的功能逻辑代码
        args参数接收传入的一行记录，若有多列，该数据从0开始表示每一列

        select fz_func2('M1001#xiaohu#S324231212','fz#M1002#S2543412432', 'S21312312412#M1003#yhd');

        输出：
            1001 xiaohu 324231212
            1002 fz 2543412432
            1003 yhd 21312312412
     */
    @Override
    public void process(Object[] args) throws HiveException {
        Object[] cols = new Object[3];

        //遍历args，获取一条数据中的每一列
        for (Object item : args) {
            //select临时数据中hive将字符串直接封装成了Text类型
            //存储在hdfs上的数据中hive将字符串直接封装成了LazyString类型
            String info = ((Text)item).toString();
            //根据#切分
            String[] is = info.split("#");
            for (String i : is) {
                if(i.startsWith("M")){
                    cols[0] = Long.parseLong(i.substring(1));
                }else if(i.startsWith("S")){
                    cols[2] = Long.parseLong(i.substring(1));
                }else {
                    cols[1] = i;
                }
            }
            forward(cols);
        }
    }


    /*
        释放一些在initialize中创建的资源
     */
    @Override
    public void close() throws HiveException {
        //若initialize方法中没有创建任何资源，这里就可以空着
    }
}
